package com.market_analysis

import java.sql.Timestamp

import com.market_analysis.bean.{AdClickCountByProvince, AdClickLog, BlackListUserWarning}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 10:43
  *                统计地区广告点击量，以及刷单黑名单
  */
object AdClickAnalysisWithBlackList {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val filePath: String = "MarketAnalysis/src/main/resources/AdClickLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val adLogStream: DataStream[AdClickLog] = fileStream.map(
      data => {
        val arr: Array[String] = data.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      }
    )
      .assignAscendingTimestamps(_.timestamp * 1000L)

    //插入一步过滤操作，并将有刷单行为的用户输出到侧输出流（黑名单报警）
    val filterBlackListUserStream: DataStream[AdClickLog] = adLogStream
      .keyBy(data => (data.userId, data.adId))
      .process(new FilterBlackListUserResult(100))


    //开窗聚合统计
    val adCountResultStream: DataStream[AdClickCountByProvince] = filterBlackListUserStream.keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5))
      .aggregate(new AdCountAgg, new AdCountWindowResult)

    adCountResultStream.print("ad")
    filterBlackListUserStream.getSideOutput(new OutputTag[BlackListUserWarning]("warning")).print("warning")

    env.execute("ad count black list job")

  }

  class AdCountAgg extends AggregateFunction[AdClickLog, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(in: AdClickLog, acc: Long): Long = acc + 1

    override def getResult(acc: Long): Long = acc

    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  class AdCountWindowResult extends WindowFunction[Long, AdClickCountByProvince, String, TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
      val windowEnd = new Timestamp(window.getEnd).toString

      out.collect(AdClickCountByProvince(windowEnd, key, input.head))
    }
  }

  //自定义KeyedProcessFunction
  class FilterBlackListUserResult(maxCount: Int) extends KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog] {
    //定义状态，保存用户对广告的点击量
    lazy val countState: ValueState[Long] = getRuntimeContext
      .getState(new ValueStateDescriptor[Long]("count", classOf[Long]))
    //每天0点定时清空状态的时间戳
    lazy val resetTimerTsState: ValueState[Long] = getRuntimeContext
      .getState(new ValueStateDescriptor[Long]("reset-ts", classOf[Long]))
    //标记当前用户是否已经进入黑名单
    lazy val isBlackState: ValueState[Boolean] = getRuntimeContext
      .getState(new ValueStateDescriptor[Boolean]("is-black", classOf[Boolean]))

    override def processElement(i: AdClickLog, context: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#Context, collector: Collector[AdClickLog]): Unit = {
      val curCount = countState.value()
      //判断只要是第一个数据来了，直接注册0点的清空状态定时器
      if (curCount == 0) {
        var ts = (context.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (1000 * 60 * 60 * 24) - 1000 * 60 * 60 * 8
        resetTimerTsState.update(ts)
        context.timerService().registerProcessingTimeTimer(ts)
      }
      //判断count值是否已经达到定义的阈值，如果超过就输出到黑名单
      if (curCount >= maxCount) {
        //判断是否已经在黑名单里，没有的话才输出侧输出流
        if (!isBlackState.value()) {
          isBlackState.update(true)
          context.output(new OutputTag[BlackListUserWarning]("warning"), BlackListUserWarning(i.userId, i.adId, "Click ad over " + maxCount + " times today"))
        }
        return
      }
      //正常情况，count + 1，然后将数据原样输出
      countState.update(curCount+1)
      collector.collect(i)
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, Long), AdClickLog, AdClickLog]#OnTimerContext, out: Collector[AdClickLog]): Unit = {
        if (timestamp==resetTimerTsState.value()){
          isBlackState.clear()
          countState.clear()
        }
    }
  }

}
